package com.jd.jobkeeper.client;

import com.jd.jobkeeper.client.http.HttpClient;
import com.jd.jobkeeper.client.http.HttpRequest;
import com.jd.jobkeeper.client.http.HttpResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.Disposable;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

public class AddressIndex {

    private static final Logger log = LoggerFactory.getLogger(AddressIndex.class);

    final JobConfig jobConfig;
    final HttpClient httpClient = HttpClient.DEFAULT;
    final Scheduler scheduler;
    Disposable refreshTask;
    ArrayList<Node> addresses;
    int addressIdx = 0;
    private volatile String version = "0";

    public AddressIndex(JobConfig jobConfig,
                        Scheduler scheduler) {
        this.jobConfig = jobConfig;
        this.scheduler = scheduler;
    }

    public Mono start() {
        String url = jobConfig.getUrl();
        if (url != null && !url.isEmpty()) {
            parseAddress(url);
            return Mono.empty();
        }

        Retry retry = Retry.backoff(3, Duration.ofSeconds(5))
                .scheduler(scheduler)
                .maxBackoff(Duration.ofSeconds(5));
        return Mono.fromCallable(this::getAddressFromIndex)
                .retryWhen(retry)
                .doOnNext(this::parseAddress)
                .doOnNext(v -> {
                    int interval = jobConfig.getRefreshAddressInterval();
                    refreshTask = scheduler.schedulePeriodically(() -> {
                        this.parseAddress(getAddressFromIndex());
                    }, interval, interval, TimeUnit.SECONDS);
                });
    }

    public List<Node> addressList() {
        ArrayList<Node> result = new ArrayList<>();
        result.addAll(addresses);
        return result;
    }

    public void stop() {
        if (refreshTask != null) {
            refreshTask.dispose();
        }
    }

    private synchronized void parseAddress(String s) {
        log.info("Parsing address list from " + s);
        if (s == null || s.isEmpty()) {
            throw new IllegalStateException("The address list is null empty!");
        }
        String serverArray[] = s.split(",");
        ArrayList<Node> result = new ArrayList<>();
        for (String str : serverArray) {
            result.add(new Node(str));
        }
        this.addresses = result;
        this.addressIdx = new Random().nextInt(addresses.size());
    }


    public synchronized Node selectNode() {
        Node node = addresses.get(addressIdx);
        addressIdx = (addressIdx + 1) % addresses.size();
        return node;
    }

    public String getAddressFromIndex() {
        log.info("Address index is get address from index {} version:{}", jobConfig.getIndex(), version);
        HttpRequest httpRequest = new HttpRequest();
        httpRequest.setUri(jobConfig.getIndex());
        httpRequest.addHeader(HttpRequest.LAST_MODIFIED, version);
        try {
            HttpResponse response = httpClient.execute(httpRequest);
            int status = response.getStatus();
            if (status == 200) {
                version = response.getLastModified();
                return response.getBody();
            } else if (status == 304) {
                return null;
            }
            throw new RuntimeException(response.getBody());
        } catch (Exception e) {
            throw new RuntimeException("Get address from " + jobConfig.getIndex() +
                    " failure", e);
        }
    }
}
